package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class ExercisePv2 {
    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        //输入数据
        SingleOutputStreamOperator<UserBehavior> stream = env.readTextFile("data/file/UserBehavior.csv").map(ads -> {
            String[] line = ads.split(",");
            return new UserBehavior(Long.valueOf(line[0]), Long.valueOf(line[1]), Integer.valueOf(line[2]), line[3], Long.valueOf(line[4])*1000);
        }).filter(x -> x.getBehavior().equals("pv"));
        WatermarkStrategy<UserBehavior> strategy = WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((user, ts) -> user.getTimestamp());
        //8.1.1	指定时间范围内网站总浏览量（PV）的统计
        mcUV1(strategy,stream);
        //mcUV12(strategy,stream);



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void mcUV12(WatermarkStrategy<UserBehavior> strategy, SingleOutputStreamOperator<UserBehavior> stream) {
        stream.assignTimestampsAndWatermarks(strategy)
                .keyBy(UserBehavior::getBehavior)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {

                    private MapState<Long, String> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //使用状态进行过滤重复的人次
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("mapState", Long.class, String.class));

                    }

                    @Override
                    public void process(String key, Context ctx, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
                        mapState.clear();
                        for (UserBehavior element : elements) {
                            mapState.put(element.getUserId(), "一小时内:" + element.getUserId());
                        }
                        List<UserBehavior> list = StreamSupport.stream(elements.spliterator(), true).collect(Collectors.toList());
                        SimpleDateFormat ft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String msg = "每小时的uv: "+ft.format(new Date(ctx.window().getStart()))+" - "+ft.format(new Date(ctx.window().getEnd()))+" result: "+list.size();
                        out.collect(msg);
                    }
                })
                .print();
    }

    private static void mcUV1(WatermarkStrategy<UserBehavior> strategy, SingleOutputStreamOperator<UserBehavior> stream) {
        stream.assignTimestampsAndWatermarks(strategy).keyBy(UserBehavior::getUserId).window(TumblingEventTimeWindows.of(Time.hours(1)))
                .process(new ProcessWindowFunction<UserBehavior, String, Long, TimeWindow>() {

                    private MapState<Long, String> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //使用状态进行过滤重复的人次
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("mapState", Long.class, String.class));

                    }

                    @Override
                    public void process(Long key, Context ctx, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
                        mapState.clear();
                        for (UserBehavior element : elements) {
                            mapState.put(element.getUserId(), "一小时内:" + element.getUserId());
                        }
                        List<UserBehavior> list = StreamSupport.stream(elements.spliterator(), true).collect(Collectors.toList());
                        SimpleDateFormat ft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String msg = "每小时的uv: "+ft.format(new Date(ctx.window().getStart()))+" - "+ft.format(new Date(ctx.window().getEnd()))+" result: "+list.size();
                        out.collect(msg);
                    }
                }).print();
    }



}
